package org.jpush.core.monitor;

import java.util.LinkedList;
import java.util.List;

import org.jpush.core.AbstractChannelStreamer.SendResult;
import org.jpush.core.AbstractChannelManager;
import org.jpush.core.Channel.PacketInfo;
import org.jpush.core.PhysicalConnection;
import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.Pointer;

public class MonitorsChannelManager extends AbstractChannelManager
{
	private List<PhysicalConnection> vectMonitorsConnections = 
		new LinkedList<PhysicalConnection>();

	private Object cs;

	public MonitorsChannelManager()
	{
	    createChannel("stats", 100, true, 10, 10);
	    createChannel("clientsIn", 1000, true, 5, 5);
	    createChannel("clientsOut", 50, true, 5, 5);

	    cs = new Object();
	}

	public void AddMonitor(PhysicalConnection monitor)
	{
		synchronized (cs)
		{
			vectMonitorsConnections.add(monitor);
		}
	}

	public boolean HandleSingleMonitor(PhysicalConnection monitorChannel)
	{
	    if (monitorChannel.checkIfUnusedByIOWorkers())
	        return true;

	    Pointer<ChannelInfo> pFoundInfo = new Pointer<ChannelInfo>();
	    Pointer<PacketInfo> pPacketInfo = new Pointer<PacketInfo>();

	    String subscriberKey = monitorChannel.getLogicalConnectionKey();
	    ChannelContext broadcastCtxt = monitorChannel.getChannelContext();

	    OutgoingPacket packet;
	    while (true)
	    {
	        packet = getNextPacket(subscriberKey, broadcastCtxt, pPacketInfo, pFoundInfo);
	        if (packet == null)
	            return false;

	        SendResult result = monitorChannel.pushPacket(packet);
	        disposePacket(pPacketInfo.getValue(), subscriberKey, pFoundInfo.getValue(), broadcastCtxt, result == SendResult.SendResult_OK);

	        if (result == SendResult.SendResult_NotOK)
	        {
	            return false;
	        }
	    }
	}

	public void HandleAllMonitors()
	{
	    synchronized (cs)
	    {
	    	PhysicalConnection monitor;
	    	for (int idx = 0; idx < vectMonitorsConnections.size();)
	    	{
	    		monitor = vectMonitorsConnections.get(idx);
		        if (HandleSingleMonitor(monitor))
		        {
		        	vectMonitorsConnections.remove(idx);
		        }
		        else
		        {
		        	idx ++;
		        }
	    	}
	    }
	}

	protected void preEncodeOutgoingPacket(OutgoingPacket pPacket)
	{
	    //Nothing to do, packet is already encoded.
	}

	protected void deleteOutgoingPacket(OutgoingPacket pPacket)
	{
		// Nothing to do;
	}
	protected void activateSubscribers(String channelName)
	{
	    HandleAllMonitors();
	}
}
